package com.william.jdbcplus.core.parallel;

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ListenableFuture;

import java.util.*;
import java.util.concurrent.Callable;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 分组并行执行引擎
 */
public final class ParallelMapperEngine extends ParallelExecutorService implements AutoCloseable, IMapperExecute {

    private static ParallelMapperEngine instance = new ParallelMapperEngine();

    private ParallelMapperEngine() {
        super(100);
    }

    private ParallelMapperEngine(int executorSize) {
        super(executorSize);
    }

    public static ParallelMapperEngine getInstance() {
        return instance;
    }

    @Override
    public <K, V> Map<K, V> executeToMap(List<K> groups, Function<K, V> mapper) {
        if (groups.isEmpty()) {
            return Collections.emptyMap();
        }

        int size = Math.max((groups.size() + this.executorSize - 1) / this.executorSize, 1);
        List<List<K>> listGroup = Lists.partition(groups, size);
        return parallelToMap(listGroup, mapper);
    }

    @Override
    public <K, V> List<V> executeToList(final List<K> groups, final Function<K, V> mapper) {
        if (groups.isEmpty()) {
            return Collections.emptyList();
        }

        int size = Math.max((groups.size() + this.executorSize - 1) / this.executorSize, 1);
        List<List<K>> listGroup = Lists.partition(groups, size);
        return parallelToList(listGroup, mapper);
    }

    private <K, V> Map<K, V> parallelToMap(final Collection<List<K>> groups, final Function<K, V> mapper) {
        Iterator<List<K>> groupsIterator = groups.iterator();
        List<K> first = groupsIterator.next();
        Collection<ListenableFuture<Map<K, V>>> resultFutures = asyncExecuteMap(Lists.newArrayList(groupsIterator), mapper);
        return getMap(syncToMap(first, mapper), resultFutures);
    }

    private <K, V> List<V> parallelToList(final Collection<List<K>> groups, final Function<K, V> mapper) {
        Iterator<List<K>> groupsIterator = groups.iterator();
        List<K> first = groupsIterator.next();
        Collection<ListenableFuture<Collection<V>>> resultFutures = asyncExecuteList(Lists.newArrayList(groupsIterator), mapper);
        return getList(syncToList(first, mapper), resultFutures);
    }

    private <K, V> Collection<ListenableFuture<Map<K, V>>> asyncExecuteMap(final List<List<K>> groups, final Function<K, V> mapper) {
        Collection<ListenableFuture<Map<K, V>>> result = new LinkedList<>();
        for (List<K> group : groups) {
            result.add(asyncToMap(group, mapper));
        }
        return result;
    }

    private <K, V> Collection<ListenableFuture<Collection<V>>> asyncExecuteList(final List<List<K>> groups, final Function<K, V> mapper) {
        Collection<ListenableFuture<Collection<V>>> result = new LinkedList<>();
        for (List<K> group : groups) {
            result.add(asyncToList(group, mapper));
        }
        return result;
    }

    private <K, V> ListenableFuture<Map<K, V>> asyncToMap(final List<K> group, final Function<K, V> mapper) {
        return executorService.submit(new Callable<Map<K, V>>() {
            @Override
            public Map<K, V> call() throws Exception {
                Map<K, V> result = new HashMap<>();
                for (K key : group) {
                    result.put(key, mapper.apply(key));
                }
                return result;
            }
        });
    }

    private <K, V> ListenableFuture<Collection<V>> asyncToList(final List<K> group, final Function<K, V> mapper) {
        return executorService.submit(new Callable<Collection<V>>() {
            @Override
            public Collection<V> call() throws Exception {
                return group.stream().map(e -> mapper.apply(e)).collect(Collectors.toList());
            }
        });
    }

    private <K, V> Map<K, V> syncToMap(final List<K> group, final Function<K, V> mapper) {
        Map<K, V> result = new HashMap<>();
        for (K key : group) {
            result.put(key, mapper.apply(key));
        }
        return result;
    }

    private <K, V> Collection<V> syncToList(final List<K> group, final Function<K, V> mapper) {
        return group.stream().map(e -> mapper.apply(e)).collect(Collectors.toList());
    }

    private <V> List<V> getList(final Collection<V> firstResults, final Collection<ListenableFuture<Collection<V>>> resultFutures) {
        List<V> result = new LinkedList<>();
        result.addAll(firstResults);
        for (ListenableFuture<Collection<V>> each : resultFutures) {
            try {
                result.addAll(each.get());
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
        return result;
    }

    private <K, V> Map<K, V> getMap(final Map<K, V> firstResults, final Collection<ListenableFuture<Map<K, V>>> resultFutures) {
        Map<K, V> result = new HashMap<>();
        result.putAll(firstResults);
        for (ListenableFuture<Map<K, V>> each : resultFutures) {
            try {
                result.putAll(each.get());
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
        return result;
    }

}
